package com.chj.utils;

import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.*;

@Slf4j
public class PulsarUtil {

    private static PulsarUtil pulsarUtil = new PulsarUtil();
    PulsarClient client = null;
    //    Producer<byte[]> producer = null;
    Consumer<byte[]> subscribe = null;


    private PulsarUtil() {
        try {
            client = PulsarClient.builder().serviceUrl("pulsar://192.168.2.180:26650").build();
        } catch (PulsarClientException e) {
            e.printStackTrace();
        }
    }

    public static PulsarUtil getInstance() {
        if (pulsarUtil == null) {
            pulsarUtil = new PulsarUtil();
        }
        return pulsarUtil;
    }

    /**
     * 订阅模式
     */
    public void subscribe() {
        try {
            subscribe = client.newConsumer().subscriptionName("subscription-test")
                    .consumerName("consumer-test").topic("persistent://sample/standalone/ns1/my-topic")
                    .subscriptionType(SubscriptionType.Shared)
                    .messageListener(new MessageListener<byte[]>() {
                        @Override
                        public void received(Consumer<byte[]> consumer, Message<byte[]> message) {
                            String s = new String(message.getData());
                            log.info("------MessageListener-------> {}", s);
                            consumer.acknowledgeAsync(message);
                        }
                    }).subscribe();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 订阅模式
     */
    public void broadcast() {
        try {
            subscribe = client.newConsumer().subscriptionName("subscription-test-01")
                    .consumerName("consumer-test-01").topic("persistent://sample/standalone/ns1/my-topic")
                    .subscriptionType(SubscriptionType.Shared)
                    .messageListener(new MessageListener<byte[]>() {
                        @Override
                        public void received(Consumer<byte[]> consumer, Message<byte[]> message) {
                            String s = new String(message.getData());
                            log.info("------MessageListener-------> {}", s);
                            consumer.acknowledgeAsync(message);
                        }
                    }).subscribe();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


}
